var amqp = require('amqplib');
const MongoClient = require('mongodb').MongoClient;
const {sleep} = require('../util.js')
const {MONGO_URL,AMQP_ADDRESS,QUEUE_SINGLE_RET_NAME,MONGO_DB_NAME,MONGO_TABLE_SINGLE_NAME} = require('../config.js')


const dao = async () => {
  let conn = await amqp.connect(AMQP_ADDRESS)
  let ch = await conn.createChannel()
  let q = await ch.assertQueue(QUEUE_SINGLE_RET_NAME, {durable: false});
  var operates = []
  var db
  process.on('SIGINT', async () => {
    await db.collection(MONGO_TABLE_SINGLE_NAME).bulkWrite(operates,{ordered:false});
    operates = []
    if (operates.length==0)
      process.exit();
  });

  while(true){
    let client = await MongoClient.connect(MONGO_URL, {useNewUrlParser: true});
    db = client.db(MONGO_DB_NAME);

    try {
      while(true){
        if (operates.length > 100){
          await db.collection(MONGO_TABLE_SINGLE_NAME).bulkWrite(operates,{ordered:false});
          operates = []
        }
        let msg = await ch.get(q.queue,{noAck:true})
        if (msg){
          let data = JSON.parse(msg.content.toString())
          // db.collection("industry_chain").updateOne({'_id': data['name'] + ' ' + data[MONGO_TABLE_SINGLE_NAME]}, { $set: data},{upsert:true} );
          operates.push( {updateOne:{filter:{'_id': data['keyword']}, update:{ $set: data},upsert:true}})
          console.log('finsh',data['keyword'])
          }
        else{
            await sleep(1000)
            continue
          }
        }
    } catch (err) {
      if (err.name == "MongoError" && err.codeName == "CursorNotFound") {
        alert( "CursorNotFound" + e.message );
        client.close();
        continue
      } else {
        console.error(err); // rethrow (*)
      }
    } finally {
      // client.close();
      return
    }
  }
}




dao()
